package com.study.chapter07;

import com.study.entity.Event;
import com.study.chapter05.source.ClickSource;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.util.*;

/**
 * @Description:    需要统计最近10秒钟内最热门的两个url链接，并且每5秒钟更新一次
 * @Author: LiuQun
 * @Date: 2022/8/5 21:59
 */
public class ProcessAllWindowTopNTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //获取数据，并分配水位线
        SingleOutputStreamOperator<Event> streamOpt = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forMonotonousTimestamps()
                                .withTimestampAssigner((data, l) -> data.timestamp)
                );

        streamOpt
                //根据url分组
                .map(data -> data.url)
                //每次统计10s内的数据，每5s滑动一次
                .windowAll(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
                .process(new ProcessAllWindowFunction<String, String, TimeWindow>() {

                    @Override
                    public void process(Context context, Iterable<String> elements, Collector<String> out) throws Exception {
                        Map<String, Long> dataMap = new HashMap<>();
                        //将数据放到map中
                        elements.forEach(url -> {
                            if (dataMap.containsKey(url)) {
                                Long num = dataMap.get(url);
                                dataMap.put(url, num + 1);
                            } else {
                                dataMap.put(url, 1L);
                            }
                        });
                        List<Tuple2<String, Long>> dataList = new ArrayList<Tuple2<String, Long>>();
                        dataMap.forEach((k, v) -> {
                            dataList.add(Tuple2.of(k, v));
                        });
                        //倒序排序
                        dataList.sort(new Comparator<Tuple2<String, Long>>() {
                            @Override
                            public int compare(Tuple2<String, Long> o1, Tuple2<String, Long> o2) {
                                return o2.f1.intValue() - o1.f1.intValue();
                            }
                        });

                        StringBuilder result = new StringBuilder();
                        result.append("========================================\n");
                        //取出前两个结果，并输出
                        for (int i = 0; i < 2; i++) {
                            Tuple2<String, Long> temp = dataList.get(i);
                            String info = "No." + (i + 1) +
                                    " url：" + temp.f0 +
                                    " num：" + temp.f1 +
                                    " 窗口结束时间：" + new Timestamp(context.window().getEnd()) + "\n";
                            result.append(info);

                        }
                        out.collect(result.toString());
                        result.append("========================================\n");

                    }
                })
                .print();
        env.execute();

    }
}
